package com.irt.rootcloud.realtime.gketl.bolt;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Tuple;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.irt.rootcloud.realtime.gketl.model.SchemaModel;
import com.irt.rootcloud.realtime.gketl.util.LogParseTool;
import com.irt.rootcloud.realtime.gketl.util.Utils;

/**
 * 将分类好的数据发送到不同的通道中
 * <p>Title:</p>
 * <p>Description:</p>
 * <p>Company:</p>
 * @author Ken.zhang
 * @date 2017-1-18 下午2:30:02
 */
public class SubSendBolt extends BaseBasicBolt{

	/**
	 * 
	 */
	private static final long serialVersionUID = -4987153371233996139L;

	//storm collector
	private OutputCollector collector;
	private Properties properties;
	private static Producer<String, String> producer;
	private String saasTopic ;
	
	public SubSendBolt(Properties properties){
		this.properties = properties;
	}
	@Override
	public void prepare(Map stormConf, TopologyContext context) {
		// TODO Auto-generated method stub
		super.prepare(stormConf, context);
		instanceProducer();
		saasTopic = properties.getProperty("saasTopic.name");
		
	}

	public void instanceProducer() {
		Properties props = new Properties();
	        //此处配置的是kafka的端口
        props.put("metadata.broker.list", properties.getProperty("kafka.broker.list"));

        //配置value的序列化类
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        //配置key的序列化类
        props.put("key.serializer.class", "kafka.serializer.StringEncoder");

        //request.required.acks
        //0, which means that the producer never waits for an acknowledgement from the broker (the same behavior as 0.7). This option provides the lowest latency but the weakest durability guarantees (some data will be lost when a server fails).
        //1, which means that the producer gets an acknowledgement after the leader replica has received the data. This option provides better durability as the client waits until the server acknowledges the request as successful (only messages that were written to the now-dead leader but not yet replicated will be lost).
        //-1, which means that the producer gets an acknowledgement after all in-sync replicas have received the data. This option provides the best durability, we guarantee that no messages will be lost as long as at least one in sync replica remains.
        props.put("request.required.acks","0");

        producer = new Producer<String, String>(new ProducerConfig(props));

	}

	@SuppressWarnings({ "rawtypes", "unchecked" })
	public void sendMessage(String message,String topicname) {
		try{
			KeyedMessage kmes = new KeyedMessage(topicname,message);
			producer.send(kmes);
		}catch(Exception e){
			e.printStackTrace();
		}
	}
	@Override
	public void cleanup() {
		// TODO Auto-generated method stub
		super.cleanup();
	}

	public void execute(Tuple input, BasicOutputCollector collector) {
		// TODO Auto-generated method stub
		try{
			String tenant = input.getStringByField(Utils.SUB_CHANNEL_TENANT);
			String devicetype = input.getStringByField(Utils.SUB_CHANNEL_DEVICETYPE);
			String dtjson = input.getStringByField(Utils.SUB_FIELDCHANNEL_BOLT);
			Map<String,String> mapStr = LogParseTool.jsonToMap(dtjson);
			List<SchemaModel> list = (List<SchemaModel>) input.getValueByField(Utils.SUB_CHANNEL_FIELDSUB);
			if(list.size()>0){
				for(SchemaModel sm:list){
					String cols = sm.getCols();
					if(cols == null) continue;
					String[] closArray = cols.split(",");
					Map<String,String> childmes = new HashMap<String,String>();
					childmes.put("sub_business_tenant", tenant);
					childmes.put("sub_business_deviceType", devicetype);
					childmes.put("sub_business_subType", sm.getTopicname());
					
					if(sm.getFilter()!=null&&!"".equals(sm.getFilter())){
						String filters = sm.getFilter();
						JSONArray objectArray = JSONArray.parseArray(filters);
						JSONObject object = objectArray.getJSONObject(0);
						String col = (String) object.get("cn");
						String val = (String) object.get("val");
						String type = (String) object.get("ty");
						String op = (String) object.get("op");
						if(filterLog(col,val,type,op,mapStr)){
							for(String key:closArray){
								childmes.put(key, mapStr.get(key));
							}
							String childjson = LogParseTool.mapToJson(childmes);
							sendMessage(childjson,saasTopic);
						}
					}else{
						for(String key:closArray){
							childmes.put(key, mapStr.get(key));
						}
						String childjson = LogParseTool.mapToJson(childmes);
						sendMessage(childjson,saasTopic);
					}
					
					
				}
			}
		}catch(Exception e){
			e.printStackTrace();
		}
	}
	// “fil”:[{cn:””,val:””,ty:””,op:””}]
//	cn:订阅列名称，必须与设备模型中列匹配；
//	val:过滤参考值；
//	ty:过滤参考值数据类型（暂只支持STRING,INT,DOUBLE）；
//	op:比较符，EQ：等于，GT:大于，GTE:大等于，LT:小于
//	LTE:小等于，NL:等于空，UNL不等于空

	private boolean filterLog(String col,String val,String type,String op ,Map<String,String> mapStr){
		boolean b = false;
		if(type.equals("INT")||type.equals("DOUBLE")){
			double a1 = 0;
			double a2 = 0;
			if(type.equals("INT")){
				String tv1 = mapStr.get(col);
				if(tv1==null){
					return false;
				}
				a1 = Integer.valueOf(tv1.toString());//值
				a2 = Integer.valueOf(val);
			}
			if(type.equals("DOUBLE")){
				String tv1 = mapStr.get(col);
				if(tv1==null){
					return false;
				}
				a1 = Double.valueOf(tv1.toString());//值
				a2 = Double.valueOf(val);
			}
			
			if(op.equals("GT")){
				if(a1>a2){
					b = true;
				}else{
					b = false;
				}
			}
			if(op.equals("LT")){
				if(a1<a2){
					b = true;
				}else{
					b = false;
				}
			}
			if(op.equals("GTE")){
				if(a1>=a2){
					b = true;
				}else{
					b = false;
				}
			}
			if(op.equals("LTE=")){
				if(a1<=a2){
					b = true;
				}else{
					b = false;
				}
			}
			if(op.equals("EQ")){
				if(a1==a2){
					b = true;
				}else{
					b = false;
				}
			}
		}else if(type.equals("STRING")){
			String c1 = mapStr.get(col);//值
			String c2 = val;
			if(op.equals("NL")){//空
				if(c1==null||c1==""){
					b = true;
				}else{
					b = false;
				}
			}
			if(op.equals("UNL")){//非空
				if(c1!=null&&c1!=""){
					b = true;
				}else{
					b = false;
				}
			}
			if(op.equals("UEQ")){//不等于
				if(c1!=null&&!c1.equals(c2)){
					b = true;
				}else{
					b = false;
				}
			}
		}
		return b;
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		
	}

}
